package cn.doitedu.rtdw.dashboard;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class JOB01_各自业务线每5分钟流量 {

    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(5000);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        // 建表，映射 kafka中 dwd层 行为日志明细宽表
        tenv.executeSql(
                " create table dwd_events_kafka (                       "+
                        "     release_channel string                             "+
                        "     ,device_type string                                 "+
                        "     ,session_id  string                                 "+
                        "     ,lat         double                                 "+
                        "     ,lng         double                                 "+
                        "     ,username    string                                 "+
                        "     ,event_id     string                                "+
                        "     ,action_time   bigint                               "+
                        "     ,properties  map<string,string>                     "+
                        "     ,user_id  bigint                                    "+
                        " 	  ,member_level_id bigint                             "+
                        "     ,password string                                    "+
                        "     ,nickname string                                    "+
                        "     ,phone string                                       "+
                        "     ,status int                                         "+
                        "     ,create_time timestamp(3)                           "+
                        "     ,icon string                                        "+
                        "     ,gender int                                         "+
                        "     ,birthday date                                      "+
                        "     ,register_city string                               "+
                        "     ,job string                                         "+
                        "     ,personalized_signature string                      "+
                        "     ,source_type int                                    "+
                        "     ,integration int                                    "+
                        "     ,growth int                                         "+
                        "     ,luckey_count int                                   "+
                        "     ,history_integration int                            "+
                        "     ,modify_time timestamp(3)                           "+
                        "     ,province string                                    "+
                        "     ,city string                                        "+
                        "     ,region string                                      "+
                        "     ,page_type string                                   "+
                        "     ,page_service string                                "+
                        "     ,page_lanmu string                                  "+
                        "     ,page_pindao string                                 "+
                        "     ,pt AS proctime()                                   "+   // 处理时间语义
                        "     ,rt AS to_timestamp_ltz(action_time,3)              "+
                        "     ,watermark for rt as rt - interval '0' second       "+   // 事件时间语义
                        " ) with (                                                "+
                        "     'connector' = 'kafka',                              "+
                        "     'topic' = 'dwd-events',                             "+
                        "     'properties.bootstrap.servers' = 'doitedu:9092',    "+
                        "     'properties.group.id' = 'goo2',                     "+
                        "     'scan.startup.mode' = 'latest-offset',              "+
                        "     'value.format'='json',                              "+
                        "     'value.json.fail-on-missing-field'='false',         "+
                        "     'value.fields-include' = 'EXCEPT_KEY'               "+
                        " )                                                       "
        );

        /**
         * tenv.executeSql("desc dwd_events_kafka").print();
         * |pt | TIMESTAMP_LTZ(3) *PROCTIME* | |                         AS PROCTIME() |                            |
         * |rt |  TIMESTAMP_LTZ(3) *ROWTIME* | | AS TO_TIMESTAMP_LTZ(`action_time`, 3) | `rt` - INTERVAL '0' SECOND |
         */


        /**
         * 创建mysql中的目标表的逻辑映射表
         */
//        tenv.executeSql(
//                " CREATE TABLE tfc_mysql (    "+
//                        "   page_service string ,                     "+
//                        "   window_start timestamp(3) ,               "+
//                        "   window_end timestamp(3) ,                 "+
//                        "   pv bigint,                                "+
//                        "   uv bigint,                                "+
//                        "   ses_cnt bigint ,                          "+
//                        "   PRIMARY KEY (page_service) NOT ENFORCED   "+
//                        " ) WITH (                                    "+
//                        " 'connector' = 'jdbc',                       "+
//                        " 'url' = 'jdbc:mysql://doitedu:3306/doit42', "+
//                        " 'table-name' = 'traffic_dashboard_1',       "+
//                        " 'username' = 'root',                        "+
//                        " 'password' = 'root'                         "+
//                        " )                                           "
//
//        );



        // 写统计sql
        // 各子业务线，每5分钟，pv数，uv数，会话数
        tenv.executeSql(
               // " INSERT INTO tfc_mysql  " +
                        "SELECT                                         "+
                        " 	page_service,                                                        "+
                        "   window_start,                                                        "+
                        " 	window_end,                                                          "+
                        //" 	count(event_id = 'page_load',1,null) as pv,                      "+
                        //" 	sum(event_id = 'page_load',1,0) as pv,                           "+
                        " 	count(1) FILTER( WHERE event_id='page_load' ) as pv,                 "+
                        " 	count(distinct user_id) as uv,                                       "+
                        " 	count(distinct session_id) as ses_cnt                                "+
                        " FROM TABLE(                                                            "+
                        "     TUMBLE(TABLE dwd_events_kafka,DESCRIPTOR(rt),INTERVAL '1' MINUTE)   "+
                        " )                                                                      "+
                        " GROUP BY                                                               "+
                        "     window_start,                                                      "+
                        " 	  window_end,                                                        "+
                        " 	  page_service                                                       "
        ).print();



    }
}
